package com.atguigu.gmall.realtime.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.atguigu.gmall.realtime.bean.{PageActionLog, PageDisplayLog, PageLog, StartLog}
import com.atguigu.gmall.realtime.utils.{KafkaUtil, OffsetUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author caodan
 * @date 2022-08-23 14:26
 * @version 1.0
 */
object BaseLogApp {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf

    conf.setMaster("local[1]").setAppName("base-log-app")

    val context = new StreamingContext(conf, Seconds(5))
    val odsBaseTopic : String = "ODS_BASE_LOG"
    val groupId : String = "ODS_BASE_LOG"

    //TODO 从redis中读取偏移量
    val offsets: Map[TopicPartition, Long] = OffsetUtil.getOffset(odsBaseTopic, groupId)


    var kafkaDStream : DStream[ConsumerRecord[String, String]]  = null
    if(offsets != null && offsets.nonEmpty){
      //redis中有记录offset
      //1. 接收Kafka数据流
      kafkaDStream =
        KafkaUtil.getKafkaDstream(odsBaseTopic,context,groupId,offsets)
    }else{
      //redis中没有记录offset
      //1. 接收Kafka数据流
      kafkaDStream =
        KafkaUtil.getKafkaDstream(odsBaseTopic,context,groupId)
    }

    //在数据转换前, 提取本次流中offset的结束点，
    var offsetRanges: Array[OffsetRange] = null  // driver
    kafkaDStream =  kafkaDStream.transform(  // 每批次执行一次
      (rdd: RDD[ConsumerRecord[String, String]]) => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    val jsonObjStream: DStream[JSONObject] = kafkaDStream.map(
      (consumerRecord: ConsumerRecord[String, String]) => {
        val str: String = consumerRecord.value()
        val value: JSONObject = JSON.parseObject(str)
        value
      }
    )

    //    jsonObjStream.print(100) 输出
    val errorTopic: String = "DWD_ERROR_TOPIC"
    val startTopic: String = "DWD_START_TOPIC"
    val pageTopic: String = "DWD_PAGE_TOPIC"
    val disPalyTopic: String = "DWD_DISPALY_TOPIC"
    val actionTopic: String = "DWD_ACTION_TOPIC"

    jsonObjStream.foreachRDD(
      (jsonObjStram: RDD[JSONObject]) => {
        jsonObjStram.foreachPartition(
          (iter: Iterator[JSONObject]) => {
            for (jsonObj <- iter) {
              //获取common数据
              val commonObj: JSONObject = jsonObj.getJSONObject("common")
              val mid: String = commonObj.getString("mid")
              val uid: String = commonObj.getString("uid")
              val ar: String = commonObj.getString("ar")
              val ch: String = commonObj.getString("ch")
              val os: String = commonObj.getString("os")
              val md: String = commonObj.getString("md")
              val vc: String = commonObj.getString("vc")
              val ba: String = commonObj.getString("ba")
              val isNew: String = commonObj.getString("is_new")
              val ts: Long = jsonObj.getLong("ts")

              val errObj: JSONObject = jsonObj.getJSONObject("error")

              //判断是否是错误日志
              if (errObj != null) {
                KafkaUtil.send(errorTopic, jsonObj.toJSONString)
              }
              //判断是否是page日志
              val pageObj: JSONObject = jsonObj.getJSONObject("page")
              if (pageObj != null) {
                //提取字段
                val pageId: String = pageObj.getString("page_id")
                val pageItem: String = pageObj.getString("item")
                val pageItemType: String = pageObj.getString("item_type")
                val lastPageId: String = pageObj.getString("last_page_id")
                val duringTime: Long = pageObj.getLong("during_time")
                val pageLog: PageLog =
                  PageLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, ts, ba)
                KafkaUtil.send(pageTopic, JSON.toJSONString(pageLog, new SerializeConfig(true)))
                //获取dispaly日志
                val disPalyArr: JSONArray = jsonObj.getJSONArray("displays")
                if (disPalyArr != null && disPalyArr.size() > 0) {
                  for (i <- 0 until disPalyArr.size()) {
                    val displayObj: JSONObject = disPalyArr.getJSONObject(i)
                    val displayType: String = displayObj.getString("display_type")
                    val displayItem: String = displayObj.getString("item")
                    val displayItemType: String = displayObj.getString("item_type")
                    val displayOrder: String = displayObj.getString("order")
                    val displayPosId: String = displayObj.getString("pos_id")
                    val displayLog: PageDisplayLog = PageDisplayLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, displayType, displayItem, displayItemType, displayOrder, displayPosId, ts)
                    KafkaUtil.send(disPalyTopic, JSON.toJSONString(displayLog, new SerializeConfig(true)))
                  }
                }
                //获取action日志
                val actionArr: JSONArray = jsonObj.getJSONArray("actions")
                if (actionArr != null && actionArr.size() > 0) {
                  for (i <- 0 until actionArr.size()) {
                    val actionObj: JSONObject = actionArr.getJSONObject(i)
                    val actionId: String = actionObj.getString("action_id")
                    val actionItem: String = actionObj.getString("item")
                    val actionItemType: String = actionObj.getString("item_type")
                    val actionTs: Long = actionObj.getLong("ts")
                    //封装Bean
                    val pageActionLog: PageActionLog =
                      PageActionLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime,actionId,actionItem,actionItemType,actionTs)
                    KafkaUtil.send(actionTopic,JSON.toJSONString(pageActionLog,new SerializeConfig(true)))
                  }
                }
              }
              //启动日志
              val startObj: JSONObject = jsonObj.getJSONObject("start")
              if (startObj != null) {
                val entry: String = startObj.getString("entry")
                val loadingTimeMs: Long = startObj.getLong("loading_time_ms")
                val openAdId: String = startObj.getString("open_ad_id")
                val openAdMs: Long = startObj.getLong("open_ad_ms")
                val openAdSkipMs: Long = startObj.getLong("open_ad_skip_ms")

                //封装Bean
                val startLog: StartLog = StartLog(mid, uid, ar, ch, isNew, md, os, vc, entry, openAdId, loadingTimeMs, openAdMs, openAdSkipMs, ts)
                //发送Kafka
                KafkaUtil.send(startTopic, JSON.toJSONString(startLog, new SerializeConfig(true)))
              }
            }
            KafkaUtil.flush()
          }
        )
        //提交偏移量到redis中
        OffsetUtil.saveOffset(odsBaseTopic,groupId,offsetRanges)
      }
    )

    context.start()

    context.awaitTermination()

  }
}


